unit uPacketIO;

interface

uses windows, classes, uclasses, unetclasses, superobject, unetwinsock2, utasket, uhashclass, uzlibex;

type

  TPacketIoHandler = class;

  TPacket = class(TRefObject)
  private
    FHeads: ISuperObject;
    FCmd: string;
    FContent: TMemoryStream;
    FStrHeads: string;
    FHeadsInValid: boolean;
    FRefIOHandler: TPacketIoHandler;
    function GetHeads: ISuperObject;
    function GetSession: ISuperObject;
    function GetStatusCode: integer;
    procedure SetStatusCode(value: integer);
    function GetCacheMd5: string;
    procedure SetCacheMd5(value: string);
    function GetSessionID: string;
    procedure SetSessionID(value: string);
    function GetSubProto: integer;
    procedure SetSubProto(value: integer);
  public
    constructor Create;
    destructor Destroy; override;
    function ParseHeads: boolean;
    procedure Compress(const LimitCompressSize: integer = 8192);
    procedure Decompress;
    procedure BindIOHandler(IO: TPacketIoHandler);
    procedure CopySession(SrcPacket: tpacket);
    property Heads: ISuperObject read GetHeads;
    property Session: ISuperObject read GetSession;
    property Cmd: string read FCmd write FCmd;
    property Content: TMemoryStream read FContent;
    property IOHandler: TPacketIoHandler read FRefIOHandler;
    property StatusCode: integer read GetStatusCode write SetStatusCode;
    property CacheMd5: string read GetCacheMd5 write SetCacheMd5;
    property SessionID: string read GetSessionID write SetSessionID;
    property SubProto: integer read GetSubProto write SetSubProto;
  end;

  TPacketHead = packed record
    szCmd: Byte;
    szHead: Cardinal;
    szContent: Cardinal;
  end;

  TOnPacket = procedure(const packet: TPacket; IOHandler: TPacketIoHandler) of object;

  TPacketDispatchItem = record
    OnPacket: TOnPacket;
    Runer: TTaskRuner;
  end;
  PPacketDispatchItem = ^TPacketDispatchItem;

  TPacketDispatchor = class
  private
    FDispatchMap: TNameHash;
  public
    constructor Create;
    destructor Destroy; override;
    procedure AddDispatchItem(const cmd: string; Pro: TOnPacket; Runer: TTaskRuner = nil);
    function DispatchPacket(const packet: TPacket; IO: TPacketIoHandler): boolean;
  end;

  TPacketReadState = (pktrdNone, pktrdHead, pktrdCmd, pktrdHeads, pktrdContent);

  TPacketIoHandler = class(TNetIoHandler)
  private
    FReadLimitSize: Cardinal;
    FReadPktHead: TPacketHead;
    FReadState: TPacketReadState;
    FReadPkt: TPacket;
    FReadCmd: AnsiString;
    FReadStrHeads: AnsiString;
    FGuid: string;
    procedure DoOnPacket;
  protected
    procedure OnBufferReady(const Buffer: pointer; count: integer); override;
    procedure OnStreamReady(const Stream: TStream); override;
    procedure OnPacket(const Packet: TPacket); virtual;
  public
    constructor Create(const ACompletionEngine: TNetCompletionEngine; const NetSkt: TNetSocket);
    destructor Destroy; override;
    procedure ReadPacket(const LimitSize: Cardinal = 1024);
    procedure WritePacket(pkt: TPacket);
    property Guild: string read FGuid;
  end;

procedure PacketDataToStream(Pkt: TPacket; Stream: TStream);

implementation

uses SysUtils;

const

  SYS_FD_SESSION    = '_SESSION';
  SYS_FD_COMPRESS   = '_COMPRESS';
  SYS_FD_STATUSCODE = '_STATUSCODE';
  SYS_FD_CLIENTCACHE = '_CLTCACHE';
  SYS_FD_SESSIONID  = '_SESSIONID';
  Sys_FD_SubProto   = '_subProto';

type

  TPacketDispatchToRuner = class(TTask)
  private
    OnPacket: TOnPacket;
    Packet: TPacket;
  protected
    procedure Execute; override;
  public
    constructor Create(const AOnPacket: TOnPacket; const APacket: TPacket);
  end;

procedure PacketDataToStream(pkt: TPacket; Stream: TStream);
var
  pktHead           : TPacketHead;
  StrHeads, StrCmd  : ansistring;
begin
  if pkt.Cmd = '' then
    raise Exception.Create('Cmd Can not be null!');
  StrHeads := ansistring(pkt.Heads.AsJSon());
  StrCmd := ansistring(pkt.FCmd);
  pktHead.szCmd := length(StrCmd);
  pkthead.szHead := htonl(length(StrHeads));
  pktHead.szContent := htonl(pkt.FContent.Size);
  Stream.WriteBuffer(pkthead, sizeof(pkthead));
  Stream.WriteBuffer(StrCmd[1], length(strCmd));
  Stream.WriteBuffer(StrHeads[1], length(StrHeads));
  Stream.CopyFrom(pkt.FContent, 0);
end;
{ TPacket }

procedure TPacket.BindIOHandler(IO: TPacketIoHandler);
begin
  if Assigned(FRefIOHandler) then
    FRefIOHandler.Release;
  FRefIOHandler := IO;
  IO.AddRef;
end;

procedure TPacket.CopySession(SrcPacket: TPacket);
begin
  Heads.O[SYS_FD_SESSION] := SrcPacket.Heads.O[SYS_FD_SESSION];
end;

constructor TPacket.Create;
begin
  FContent := TMemoryStream.Create;
end;

procedure TPacket.Decompress;
var
  mem               : TMemoryStream;
begin
  if not heads.B[SYS_FD_COMPRESS] then
    Exit;
  heads.B[SYS_FD_COMPRESS] := false;
  mem := TMemoryStream.Create;
  ZDecompressStream(Content, mem);
  FContent.Free;
  FContent := mem;
end;

destructor TPacket.Destroy;
begin
  if Assigned(FContent) then
    FContent.Free;
  if Assigned(FRefIOHandler) then
    FRefIOHandler.Release;
  inherited;
end;

function TPacket.GetHeads: ISuperObject;
begin
  if not Assigned(FHeads) then
  begin
    FHeads := SO(FStrHeads);
    if not ObjectIsType(FHeads, stObject) then
    begin
      FHeadsInValid := true;
      FHeads := SO();
    end
    else
      FStrHeads := '';
  end;
  Result := FHeads;
end;

function TPacket.GetSession: ISuperObject;
begin
  result := Heads.O[SYS_FD_SESSION];
  if Result = nil then
  begin
    Result := so;
    Heads.O[SYS_FD_SESSION] := result;
  end;
end;

function TPacket.ParseHeads: boolean;
begin
  GetHeads;
  Result := not FHeadsInValid;
end;

procedure TPacket.Compress(const LimitCompressSize: integer);
var
  mem               : TMemoryStream;
begin
  if (content.Size >= LimitCompressSize) and (not heads.B[SYS_FD_COMPRESS]) then
  begin
    mem := TMemoryStream.Create;
    content.Position := 0;
    ZCompressStream(Content, mem);
    FContent.Free;
    FContent := mem;
    heads.B[SYS_FD_COMPRESS] := true;
  end;
end;

function TPacket.GetStatusCode: integer;
begin
  Result := Heads.I[SYS_FD_STATUSCODE];
end;

procedure TPacket.SetStatusCode(value: integer);
begin
  Heads.I[SYS_FD_STATUSCODE] := value;
end;

function TPacket.GetCacheMd5: string;
begin
  Result := heads.S[SYS_FD_CLIENTCACHE];
end;

procedure TPacket.SetCacheMd5(value: string);
begin
  heads.S[SYS_FD_CLIENTCACHE] := value;
end;

function TPacket.GetSessionID: string;
begin
  Result := Session.S[SYS_FD_SESSIONID];
end;

procedure TPacket.SetSessionID(value: string);
begin
  Session.S[SYS_FD_SESSIONID] := value;
end;

function TPacket.GetSubProto: integer;
begin
  result := heads.I[Sys_FD_SubProto];
end;

procedure TPacket.SetSubProto(value: integer);
begin
  Heads.I[Sys_FD_SubProto] := value;
end;

{ TPacketIoHandler }

constructor TPacketIoHandler.Create(
  const ACompletionEngine: TNetCompletionEngine; const NetSkt: TNetSocket);
var
  G                 : TGUID;
begin
  inherited Create(ACompletionEngine, NetSkt);
  CreateGUID(G);
  FGuid := GUIDToString(G);
end;

destructor TPacketIoHandler.Destroy;
begin
  if Assigned(FReadPkt) then
    FReadPkt.Release;
  inherited;
end;

procedure TPacketIoHandler.DoOnPacket;
var
  pkt               : TPacket;
begin
  FReadState := pktrdNone;
  pkt := FReadPkt;
  FReadPkt := nil;
  try
    OnPacket(pkt);
  finally
    pkt.Release;
  end;
end;

procedure TPacketIoHandler.OnBufferReady(const Buffer: pointer;
  count: integer);
var
  tSize             : Cardinal;
begin
  case FReadState of
    pktrdHead:
      begin
        FReadPktHead.szHead := ntohl(FReadPktHead.szHead);
        FReadPktHead.szContent := ntohl(FReadPktHead.szContent);
        tSize := FReadPktHead.szCmd + FReadPktHead.szHead + FReadPktHead.szContent;
        if (FReadPktHead.szCmd = 0)
          or
          (FReadPktHead.szHead = 0)
          or
          ((FReadLimitSize > 0) and (tSize > FReadLimitSize)) then
        begin          
          Close
        end
        else
        begin
          FReadState := pktrdCmd;
          SetLength(FReadCmd, FReadPktHead.szCmd);
          ReadBuffer(@FReadCmd[1], FReadPktHead.szCmd);
        end;
      end;
    pktrdCmd:
      begin
        FReadPkt.FCmd := string(FReadCmd);
        FReadCmd := '';
        FReadState := pktrdHeads;
        setlength(FReadStrHeads, FReadPktHead.szHead);
        ReadBuffer(@FReadStrHeads[1], FReadPktHead.szHead);
        ;
      end;
    pktrdHeads:
      begin
        FReadPkt.FStrHeads := string(FReadStrHeads);
        FReadStrHeads := '';
        if FReadPktHead.szContent > 0 then
        begin
          FReadState := pktrdContent;
          ReadStream(FReadPkt.FContent, FReadPktHead.szContent);
        end
        else
          DoOnPacket;
      end;
  end;
end;

procedure TPacketIoHandler.OnPacket(const Packet: TPacket);
begin

end;

procedure TPacketIoHandler.OnStreamReady(const Stream: TStream);
begin
  DoOnPacket;
end;

procedure TPacketIoHandler.ReadPacket;
begin
  if FReadState <> pktrdNone then
    raise Exception.Create('Read Packet State Error!');
  FReadPkt := TPacket.Create;
  FReadLimitSize := LimitSize;
  FReadState := pktrdHead;
  ReadBuffer(@FReadPktHead, sizeof(FReadPktHead));
end;

procedure TPacketIoHandler.WritePacket(pkt: TPacket);
var
  mem               : TMemoryStream;
begin
  Mem := TMemoryStream.Create;
  PacketDataToStream(pkt, mem);
  WriteStream(Mem);
end;

{ TPacketDispatchor }

procedure TPacketDispatchor.AddDispatchItem(const cmd: string;
  Pro: TOnPacket; Runer: TTaskRuner);
var
  item              : PPacketDispatchItem;
begin
  if FDispatchMap.ValueOf(cmd) <> nil then
    raise Exception.Create('cmd has add to dispatch map!');
  new(item);
  item.OnPacket := Pro;
  item.Runer := Runer;
  FDispatchMap.Add(cmd, item);
end;

constructor TPacketDispatchor.Create;
begin
  FDispatchMap := TNameHash.Create(1024);
end;

destructor TPacketDispatchor.Destroy;
begin
  with FDispatchMap.GetIterator do
  begin
    first;
    while not Eof do
    begin
      Dispose(PPacketDispatchItem(Current.Data));
      next;
    end;
  end;
  FDispatchMap.Free;
  inherited;
end;

function TPacketDispatchor.DispatchPacket(const packet: TPacket; IO: TPacketIoHandler): boolean;
var
  item              : PPacketDispatchItem;
begin
  Result := false;
  item := FDispatchMap.ValueOf(packet.Cmd);
  if Assigned(item) then
  begin
    Result := true;
    if Assigned(item.Runer) then
    begin
      packet.BindIOHandler(IO);
      TPacketDispatchToRuner.Create(item.OnPacket, packet).Run(item.Runer);
    end
    else
      item.OnPacket(packet, IO);
  end;
end;

{ TPacketDispatchToRuner }

constructor TPacketDispatchToRuner.Create(const AOnPacket: TOnPacket;
  const APacket: TPacket);
begin
  OnPacket := AOnPacket;
  Packet := APacket;
  BindRefObject(Packet);
end;

procedure TPacketDispatchToRuner.Execute;
begin
  OnPacket(Packet, packet.IOHandler);
end;

end.

